package rtc

import (
	"a-common/utils/getid"
	"bufio"
	"errors"
	"fmt"
	"strconv"
	"sync"
	"time"

	"github.com/gofiber/fiber/v2"
	"github.com/valyala/fasthttp"
)

var (
	rtcSseControllerTemp *RtcSseController
	rtcSseControllerOnce sync.Once
)

type RtcSseController struct {
	online int64
	SseMap sync.Map
}

func NewRtcSseController() *RtcSseController {
	rtcSseControllerOnce.Do(func() {
		rtcSseControllerTemp = &RtcSseController{}
	})
	return rtcSseControllerTemp
}

// SseHandler SSE 通讯
func (m *RtcSseController) SseHandler(c *fiber.Ctx) error {
	cid := c.Query("id")
	if cid == "" {
		fmt.Println("id is nil")
		return errors.New("id is nil")
	}

	sessionId := getid.New().GetNanoId()
	if sessionId == "" {
		fmt.Println("sessionId is nil")
		return errors.New("sessionId is nil")
	}
	var sse SseClient
	sse.Create(cid)
	m.online++
	fmt.Printf("New sse Session=%s, UserId=%s, onLine=%d \n", sessionId, sse.GetUserId(), m.online)

	i := 0
	c.Set("Content-Type", "text/event-stream")
	c.Set("Cache-Control", "no-cache")
	c.Set("Connection", "keep-alive")
	c.Set("Transfer-Encoding", "chunked")
	c.Status(fiber.StatusOK).Context().SetBodyStreamWriter(fasthttp.StreamWriter(func(w *bufio.Writer) {
		m.SseMap.Store(sessionId, &sse)
		// 创建一个定时器
		ticker := time.NewTicker(3 * time.Second)
		defer func() {
			ticker.Stop()
			m.online--
			// 删除会话
			m.SseMap.Delete(sessionId)
			fmt.Printf("Lost sse Session=%s, UserId=%s onLine=%d \n", sessionId, sse.GetUserId(), m.online)
		}()
		for {
			select {
			case value, ok := <-sse.GetData():
				if !ok {
					return
				} else {
					i++
					fmt.Fprintf(w, "data: %d - %s\n\n", i, string(value))
					err := w.Flush()
					if err != nil {
						return
					}
				}
			case <-ticker.C:
				i++
				msg := fmt.Sprintf("data: %d - Session=%s, ClientId=%s, addr=%p\n\n", i, sessionId, sse.GetUserId(), &sse)
				fmt.Fprint(w, string(msg))
				err := w.Flush()
				if err != nil {
					// Refreshing page in web browser will establish a new
					// SSE connection, but only (the last) one is alive, so
					// dead connections must be closed here.
					// fmt.Printf("Error while flushing: %v. Closing http connection.\n", err)
					return
				}
			}
		}
	}))
	return nil
}

// SendTestMsg 发送测试消息
func (m *RtcSseController) SendTestMsg(c *fiber.Ctx) error {
	cId := c.Params("userid")
	msg := c.Params("msg")
	i := m.SendSseMsg(cId, []byte(msg+"\n"))
	return c.SendString("消息已投递:" + strconv.FormatInt(i, 10))
}

func (m *RtcSseController) SendSseMsg(clientId string, msg []byte) int64 {
	var i int64
	// 遍历 sync.Map
	m.SseMap.Range(func(key, value any) bool {
		_, ok := key.(string)
		if !ok {
			return true
		}
		v, ok := value.(*SseClient)
		if !ok {
			return true
		}
		if clientId == "0" {
			// 广播
			v.GetData() <- msg
			i++
		} else {
			// p2p
			if v.GetUserId() == clientId {
				i++
				v.GetData() <- msg
			}
		}
		// 如果返回 false，则会停止遍历
		// 在这里我们总是返回 true，以遍历所有的键值对
		return true
	})
	return i
}

func (m *RtcSseController) Ssetest(c *fiber.Ctx) error {
	c.Response().Header.Set("Content-Type", "text/html")
	return c.SendString(`<!DOCTYPE html>
	<html>
	<body>
	
	<h1>SSE Messages</h1>
	<div id="result"></div>
	
	<script>
		if (typeof(EventSource) === 'undefined') {
			document.getElementById("result").innerHTML = "Sorry, your browser does not support server-sent events...";
		}
	
		const source = new EventSource("http://127.0.0.1:8080/dev/rtc/sse/1");
	
		source.addEventListener('open', (event) => {
			document.getElementById("result").innerHTML += 'connected<br>';
		})
	
		source.addEventListener('message', (event) => {
			document.getElementById("result").innerHTML += event.data + "<br>";
		})
	
		source.addEventListener('error', (event) => {
			document.getElementById("result").innerHTML += 'connection lost<br>';
	
			source.close()
	
			document.getElementById("result").innerHTML += 'closed<br>';
		})
	</script>
	
	</body>
	</html>`)
}
